/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2013_06_06			*/
/*	File:		MrRed.cpp			*/
/********************************************************/

#include "MrRed.h"

CMrUdp g_CUdp;
bool g_bDebug = false;

int main(int argc, char* argv[])
{
	int iCount = 0;
	int iIndex = 0;
	int iReturn = 0;
	int iLength = 0;
	int iThread = 0;
	CMrThread* pCThread = NULL;
	char szPort[MR_VALUE] = { 0 };
	char szConf[MR_VALUE] = "MrRed.conf";
	UCHAR uszPacket[MR_BIGPACKET] = { 0 };

	struct option stOptions[] = {
		{"conf", 1, 0, 'c'},
		{"pool", 1, 0, 'p'},
		{"debug", 0, 0, 'd'},
		{"help", 0, 0, 'h'},
		{"version", 0, 0, 'v'},
		{0, 0, 0, 0},
	};

	while (true) {
		iReturn = getopt_long(argc, argv, "c:p:dhv", stOptions, &iIndex);
		if (iReturn < 0) {
			break;
		}
		switch (iReturn) {
			case 'c':
				strncpy(szConf, optarg, MR_VALUE);
				break;
			case 'p':
				iThread = atoi(optarg);
				break;
			case 'd':
				g_bDebug = true;
				break;
			case 'h':
				MrTkHelp(argv[0], MR_RD_VERSION);
				return MR_SUCCESS;
			case 'v':
				MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
				MR_PRINT("%s Build on %s %s.\n", MR_RD_VERSION, __DATE__, __TIME__);
				return MR_SUCCESS;
			default:
				return MR_SUCCESS;
		}
	}

	signal(SIGSEGV, MrTkQuit);
	signal(SIGTERM, MrTkQuit);

	/* config */
	if (true == g_bDebug) {
		MR_DEBUG("Initialize config from %s\n", szConf);
	}
	if (MR_SUCCESS == MrTkConfig(szConf, MR_RD_CONFIG, szPort)) {
		if (true == g_bDebug) {
			MR_DEBUG("Port: %s\n", szPort);
		}
	}
	else {
		return MR_FAILURE;
	}

	/* fork */
	if (true != g_bDebug) {
		MrFork();
	}

	/* listen */
	if (true == g_bDebug) {
		MR_DEBUG("MrRed will listen...\n");
	}
	if (MR_SUCCESS == g_CUdp.MrNetInitRecv(NULL, szPort)) {
		if (iThread > 0) {
			/* pool */
			pCThread = new CMrThread[iThread];
			for (iCount = 0; iCount < iThread; iCount ++) {
				(pCThread + iCount)->MrThreadAttrInit();
				(pCThread + iCount)->MrThreadSetStartRoutine(MrRedThread);
				(pCThread + iCount)->MrThreadCreate(NULL);
			}
			for (iCount = 0; iCount < iThread; iCount ++) {
				(pCThread + iCount)->MrThreadJoin(NULL);
			}
		}
		else {
			/* fork */
			while (true) {
				if ((iLength = g_CUdp.MrNetRecv(uszPacket, MR_BIGPACKET)) > 0) {
					MrTkResponse(g_CUdp);
					MrRedFork(g_CUdp.MrUdpGetAddr(), uszPacket, iLength);
				}
			}
		}

		g_CUdp.MrNetClose();
	}
	else {
		MR_ERROR("Failed to MrNetInitRecv\n");
	}

	return MR_SUCCESS;
}

void MrRedFork(void* pAddr, UCHAR* puszPacket, int iLength)
{
	struct sockaddr_in stAddr;
	UCHAR uszPacket[MR_BIGPACKET] = { 0 };

	signal(SIGCHLD, SIG_IGN);
	memcpy(uszPacket, puszPacket, iLength);
	memcpy(&stAddr, pAddr, sizeof(struct sockaddr_in));

	if (true != g_bDebug) {
		switch (fork()) {
			case MR_SUCCESS:
				/* MrRedChild */
				MrRedProcess(&stAddr, uszPacket, iLength);
				exit(MR_SUCCESS);
				break;
			case MR_FAILURE:
				/* MrRedError */
				break;
			default:
				/* MrRedParent */
				break;
		}
	}
	else {
		/* MrRedChild */
		MrRedProcess(&stAddr, uszPacket, iLength);
	}

	return;
}

void MrRedProcess(struct sockaddr_in* pstAddr, UCHAR* puszPacket, int iLength)
{
	CMrUdp CUdp;
	CMrTime CTime;

	if (true == g_bDebug) {
		MR_PRINT("\n");
		CTime.MrTimeOn();
		MR_DEBUG("Process[%d] start\n", getpid());
	}

	if (MR_SUCCESS == CUdp.MrUdpConnect(pstAddr)) {
		MrRedCore(CUdp, puszPacket, iLength);
		CUdp.MrNetClose();
	}

	if (true == g_bDebug) {
		MR_DEBUG("Process[%d] end Use time ", getpid());
		CTime.MrTimeOff();
		CTime.MrTimeDisplay();
	}

	return;
}

void* MrRedThread(void* pNULL)
{
	CMrUdp CUdp;
	CMrTime CTime;
	int iLength = 0;
	UCHAR uszPacket[MR_BIGPACKET] = { 0 };

	CUdp.MrNetSetSocket(g_CUdp.MrNetGetSocket());
	while (true) {
		if ((iLength = CUdp.MrNetRecv(uszPacket, MR_BIGPACKET)) > 0) {
			if (true == g_bDebug) {
				MR_PRINT("\n");
				CTime.MrTimeOn();
				MR_DEBUG("Thread[%lu] start\n", pthread_self());
			}

			MrTkResponse(CUdp);
			MrRedCore(CUdp, uszPacket, iLength);

			if (true == g_bDebug) {
				MR_DEBUG("Thread[%lu] end Use time ", pthread_self());
				CTime.MrTimeOff();
				CTime.MrTimeDisplay();
			}
		}
	}

	pthread_exit(NULL);
}

void MrRedCore(CMrUdp& rCUdp, UCHAR* puszPacket, int iLength)
{
	CMrSet CHub;
	CMrPro CPro;
	USHORT usKv = 0;
	char* pszAdp = NULL;
	MRPARA_S stIn = { 0 };
	MRPARA_S stOut = { 0 };
	MRPROKV_S stKv[MR_TK_PAIR];
	UCHAR ucResult = MR_DEF_SUCC;
	UCHAR uszPacket[MR_PACKET] = { 0 };
	UCHAR (*pFunction)(MRPARA_S&, MRPARA_S&) = NULL;

	/* head */
	CPro.MrProParse(puszPacket, iLength);
	if (true == g_bDebug) {
		MR_DEBUG("<Request>:\n");
		CPro.MrProDisplay();
	}

	/* body */
	if (CPro.MrProGetLength() > 0) {
		memset(stKv, '\0', sizeof(MRPROKV_S) * MR_TK_PAIR);
		MrProKvInput(stKv, CPro.MrProGetTotal(), CPro.MrProGetBody(), CPro.MrProGetLength());

		if (true == g_bDebug) {
			MR_DEBUG("MrSetInit total[%d] and size[%d]\n", CPro.MrProGetTotal(), MR_URL);
		}
		if (MR_SUCCESS == CHub.MrSetInit(CPro.MrProGetTotal(), MR_URL)) {
			/* para */
			if (true == g_bDebug) {
				MR_DEBUG("Parse into ");
			}
			stIn.m_uiSize = CPro.MrProGetTotal();
			while (usKv < CPro.MrProGetTotal()) {
				switch (stKv[usKv].m_ucKey) {
					case MR_MR_ADP:
						if (true == g_bDebug) {
							MR_PRINT("MR_MR_ADP ");
						}
						pszAdp = (char*)stKv[usKv].m_puszValue;
						stIn.m_uiSize -= 1;
						break;
					case MR_MR_GLB:
						if (true == g_bDebug) {
							MR_PRINT("MR_MR_GLB ");
						}
						stIn.m_pPara = (void*)&stKv[usKv];
						break;
					case MR_MR_LCL:
						if (true == g_bDebug) {
							MR_PRINT("MR_MR_LCL ");
						}
						CHub.MrSetIndexPut(CHub.MrSetGetCount(), stKv[usKv].m_puszValue);
						ucResult = MrRedLocal(&stKv[usKv]);
						break;
					default:
						break;
				}
				if (MR_DEF_SUCC != ucResult) {
					if (true == g_bDebug) {
						MR_ERROR("Failed to MrRedLocal\n");
					}
					break;
				}
				usKv += 1;
			}
			if (true == g_bDebug) {
				MR_PRINT("\n");
			}

			/* type */
			if (MR_DEF_SUCC == ucResult) {
				if (true == g_bDebug) {
					MR_DEBUG("Type: 0x%02x, ", CPro.MrProGetType());
					MR_PRINT("Verb: 0x%02x, ", CPro.MrProGetVerb());
					MR_PRINT("Attr: 0x%02x\n", CPro.MrProGetAttr());
				}
				switch (CPro.MrProGetType()) {
					default:
						#ifdef _MR_TEST_
						switch (CPro.MrProGetAttr()) {
							case MR_MR_RD1:
								pFunction = MrRedTest;
								break;
							case MR_MR_RD2:
								pFunction = MrRedTest;
								break;
							default:
								break;
						}
						#else
						ucResult = MR_DEF_TYPE;
						#endif
						break;
				}
			}

			/* run */
			if (MR_DEF_SUCC == ucResult) {
				stOut.m_uiSize = MR_PRO_1;
				if (true == g_bDebug) {
					MR_DEBUG("MrProKvMalloc %d * %d\n", stOut.m_uiSize, MR_SOCK_BUF);
				}
				if (NULL != (stOut.m_pPara = MrProKvMalloc(stOut.m_uiSize, MR_SOCK_BUF))) {
					if (true == g_bDebug) {
						MR_DEBUG("CJOB run\n");
					}
					ucResult = (*pFunction)(stIn, stOut);
					if (true == g_bDebug) {
						MR_DEBUG("CJOB end\n");
					}

					if (MR_DEF_SUCC == ucResult) {
						if (true == g_bDebug) {
							MR_DEBUG("CJOB return: success\n");
						}
						if (((MRPROKV_S*)stOut.m_pPara)->m_iLength > 0) {
							if (true == g_bDebug) {
								MR_DEBUG("Will MrTkOutput packet\n");
							}
							ucResult = MrTkOutput(CHub, stOut, CPro, &uszPacket[MR_PRO_HEAD]);
						}
					}
					else {
						if (true == g_bDebug) {
							MR_DEBUG("CJOB return: failure\n");
						}
					}
	
					if (true == g_bDebug) {
						MR_DEBUG("MrProKvFree\n");
					}
					MrProKvFree((MRPROKV_S**)&stOut.m_pPara, stOut.m_uiSize);
				}
				else {
					MR_ERROR("Failed to MrProKvMalloc\n");
					ucResult = MR_DEF_MALC;
				}
			}

			/* free */
			usKv = 0;
			while (usKv < stIn.m_uiSize) {
				if (MR_MR_LCL == stKv[usKv].m_ucKey) {
					if (stKv[usKv].m_iLength > 0) {
						MrFree((void**)&stKv[usKv].m_puszValue);
					}
					else {
						break;
					}
				}
				usKv += 1;
			}

			if (true == g_bDebug) {
				MR_DEBUG("MrSetDestroy\n");
			}
			CHub.MrSetDestroy();
		}
		else {
			MR_ERROR("Failed to MrSetInit\n");
			ucResult = MR_DEF_MALC;
		}
	}
	else {
		MR_ERROR("Failed to MrProGetLength\n");
		ucResult = MR_DEF_LENG;
	}

	/* resp */
	if (MR_DEF_SUCC != ucResult) {
		CPro.MrProPutTotal(MR_PRO_0);
		CPro.MrProPutLength(MR_PRO_0);
	}
	CPro.MrProPutBody(NULL);
	CPro.MrProPutResult(ucResult);
	CPro.MrProPutResponse(MR_PRO_N);
	CPro.MrProCreate(uszPacket, MR_PACKET);
	if (true == g_bDebug) {
		MR_DEBUG("<Response>:\n");
		CPro.MrProDisplay();
	}

	if (NULL != pszAdp) {
		if (true == g_bDebug) {
			MR_DEBUG("Response to MrAdapter\n");
		}
		rCUdp.MrUdpSetAddr(pszAdp);
	}
	else {
		if (true == g_bDebug) {
			MR_DEBUG("Response to MrProcess\n");
		}
	}
	if (true == g_bDebug) {
		MR_DEBUG("Will MrNetSend packet\n");
	}
	rCUdp.MrNetSend(uszPacket, MR_PRO_HEAD + CPro.MrProGetLength());

	return;
}

UCHAR MrRedLocal(MRPROKV_S* pstKv)
{
	CMrPro CPro;
	CMrTcp CTcp;
	int iLength = 0;
	UCHAR* puszBody = NULL;
	MRPROKV_S stKv[MR_MR_VAL];
	UCHAR ucReturn = MR_DEF_SUCC;
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };
	UCHAR uszPacket[MR_PACKET] = { 0 };
	char* pszUrl = (char*)pstKv->m_puszValue;
	MRPROHEAD_S stHead = {
		MR_PRO_FLAG,
		MR_PRO_1,
		MR_PRO_DATA,
		MR_PRO_0,
		MR_PRO_Y,
		MR_PRO_0,
		MR_DEF_EHUB,
		MR_DEF_SELT,
		MR_PRO_0,
		MR_DEF_SUCC,
		MR_MR_KEY,
		MR_PRO_0
	};

	pstKv->m_ucKey = MR_MR_KEY;
	pstKv->m_iLength -= MR_URL;
	pstKv->m_puszValue += MR_URL;

	CPro.MrProPutBody(NULL);
	CPro.MrProInduce(&stHead);
	CPro.MrProPutLength(MrProKvOutput(pstKv, MR_MR_KEY, &uszPacket[MR_PRO_HEAD], MR_PACKET - MR_PRO_HEAD));
	CPro.MrProCreate(uszPacket, MR_PACKET);

	if (MR_SUCCESS == CTcp.MrTcpConnect(pszUrl)) {
		CTcp.MrTcpSetLinger(0);
		CTcp.MrTcpSetNoDelay();
		if (CTcp.MrNetSend(uszPacket, MR_PRO_HEAD + CPro.MrProGetLength()) > 0) {
			if (MR_PRO_HEAD == CTcp.MrNetRecv(uszHead, MR_PRO_HEAD)) {
				CPro.MrProParse(uszHead, MR_PRO_HEAD);
				if ((iLength = CPro.MrProGetLength()) > 0) {
					if (NULL != (puszBody = (UCHAR*)MrMalloc(sizeof(UCHAR) * iLength))) {
						if (iLength == CTcp.MrNetRecvAll(puszBody, iLength)) {
							memset(stKv, '\0', sizeof(MRPROKV_S) * MR_MR_VAL);
							MrProKvInput(stKv, CPro.MrProGetTotal(), puszBody, iLength);
							if (NULL != (pstKv->m_puszValue = (UCHAR*)MrMalloc(sizeof(UCHAR) * stKv[1].m_iLength))) {
								pstKv->m_ucKey = MR_MR_LCL;
								pstKv->m_iLength = stKv[1].m_iLength;
								memcpy(pstKv->m_puszValue, stKv[1].m_puszValue, stKv[1].m_iLength);
							}
							else {
								MR_ERROR("Failed to MrMalloc\n");
								ucReturn = MR_DEF_MALC;
							}
						}
						else {
							MR_ERROR("Failed to MrNetRecvAll\n");
							ucReturn = MR_DEF_RECV;
						}
						MrFree((void**)&puszBody);
					}
					else {
						MR_ERROR("Failed to MrMalloc\n");
						ucReturn = MR_DEF_MALC;
					}
				}
				else {
					MR_ERROR("Failed to MrProGetLength\n");
					ucReturn = MR_DEF_LENG;
				}
			}
			else {
				MR_ERROR("Failed to MrNetRecv\n");
				ucReturn = MR_DEF_RECV;
			}
		}
		else {
			MR_ERROR("Failed to MrNetSend\n");
			ucReturn = MR_DEF_SEND;
		}

		CTcp.MrNetClose();
	}
	else {
		MR_ERROR("Failed to MrTcpConnect\n");
		ucReturn = MR_DEF_CONN;
	}

	if (MR_DEF_SUCC != ucReturn) {
		pstKv->m_iLength = MR_PRO_0;
	}

	return ucReturn;
}

#ifdef _MR_TEST_
UCHAR MrRedTest(MRPARA_S& stIn, MRPARA_S& stOut)
{
	UINT uiCount = 0;
	char* pszCur = NULL;
	char szLcl[MR_BUFFER] = { 0 };
	char szGlb[MR_BUFFER] = { 0 };
	MRPROKV_S* pstIn = (MRPROKV_S*)stIn.m_pPara;
	MRPROKV_S* pstOut = (MRPROKV_S*)stOut.m_pPara;

	pszCur = (char*)pstOut->m_puszValue;
	sprintf(pszCur, "Red(Proc: %d): ", getpid());
	for (uiCount = 0; uiCount < stIn.m_uiSize; uiCount ++) {
		switch ((pstIn + uiCount)->m_ucKey) {
			case MR_MR_GLB:
				memcpy(szGlb, (pstIn + uiCount)->m_puszValue, (pstIn + uiCount)->m_iLength);
				sprintf(pszCur + strlen(pszCur), "(Glb: %s)\n", szGlb);
				break;
			case MR_MR_LCL:
				memcpy(szLcl, (pstIn + uiCount)->m_puszValue, (pstIn + uiCount)->m_iLength);
				sprintf(pszCur + strlen(pszCur), "\tLcl: %s", szLcl);
				break;
			default:
				break;
		}
	}
	pstOut->m_iLength = strlen(pszCur);

	return MR_DEF_SUCC;
}
#endif
